package com.atguigu.flink.datastream.transform;

import com.atguigu.flink.func.ClickSource;
import com.atguigu.flink.pojo.Event;
import com.atguigu.flink.pojo.Wordcount;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author WEIYUNHUI
 * @date 2023/6/13 14:06
 * <p>
 * 用户自定义普通函数:
 * MapFunction
 * FilterFunction
 * FlatMapFunction
 * ReduceFunction
 * ....
 * <p>
 * 自定义函数的方式:
 * 1. 自定义类，实现函数接口
 * 2. 匿名内部类的方式
 * 3. Lambda
 */
public class Flink04_UserDefineFunction {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //读取socket文本数据，将数据封装成Event，打印输出
        DataStreamSource<String> ds = env.socketTextStream("hadoop102", 8888);
        //Lambada写法
        /*
        SingleOutputStreamOperator<Event> eventDs = ds.map(
                line -> {
                    String[] fields = line.split(",");
                    return new Event(fields[0].trim(), fields[1].trim(), Long.valueOf(fields[2].trim()));
                }
        );
        eventDs.print();

         */

        //匿名内部类方式
        /*
        SingleOutputStreamOperator<Event> eventDs1 = ds.map(

                new MapFunction<String, Event>() {
                    @Override
                    public Event map(String line) throws Exception {
                        String[] fields = line.split(",");
                        return new Event(fields[0].trim(), fields[1].trim(), Long.valueOf(fields[2].trim()));
                    }
                }
        );

        ventDs1.print();

         */
        ds.map(
                new MyMapFunction(" ")
        ).print();

        //自定义类

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    public static class MyMapFunction implements MapFunction<String, Event> {

        private String seperator;

        public MyMapFunction(String seperator) {
            this.seperator = seperator;
        }

        @Override
        public Event map(String line) throws Exception {
            String[] fields = line.split(seperator);
            return new Event(fields[0].trim(), fields[1].trim(), Long.valueOf(fields[2].trim()));
        }
    }
}
